package com.example.kafkaspringboot.producer;

import com.example.kafkaspringboot.config.KafkaSpringBootConfig;
import lombok.extern.slf4j.Slf4j;
import org.omg.SendingContext.RunTime;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.HashMap;

/**
 * @author He Changjie on 2021/6/30
 */
@RestController
@RequestMapping
@Slf4j
public class Producer1 {

    @Autowired
    private KafkaTemplate kafkaTemplate;

    /**
     * 发送消息
     *
     * 示例需要注释配置属性 spring.kafka.producer.transaction-id-prefix
     */
    @GetMapping("/send/{input}")
    public String sendToKafka(@PathVariable String input){
        this.kafkaTemplate.send(KafkaSpringBootConfig.TOPIC, input);
        return "send success! " + input;
    }

    /**
     * 发送消息 通过编码开启事务
     *
     * 示例需要开启配置属性 spring.kafka.producer.transaction-id-prefix
     */
    @GetMapping("/send2/{input}")
    public String sendToKafka2(@PathVariable String input){
        kafkaTemplate.executeInTransaction(e -> {
            e.send(KafkaSpringBootConfig.TOPIC, input);
            if("error".equals(input)){
                throw new RuntimeException("input is error");
            }
            e.send(KafkaSpringBootConfig.TOPIC, input + " two");
           return true;
        });
        return "send success! " + input;
    }

    /**
     * 发送消息 通过注解开启事务
     *
     * 示例需要开启配置属性 spring.kafka.producer.transaction-id-prefix
     */
    @GetMapping("/send3/{input}")
    @Transactional(rollbackFor = RuntimeException.class)
    public String sendToKafka3(@PathVariable String input){
        kafkaTemplate.send(KafkaSpringBootConfig.TOPIC, input);
        if("error".equals(input)){
            throw new RuntimeException("input is error");
        }
        kafkaTemplate.send(KafkaSpringBootConfig.TOPIC, input + " two");
        return "send success! " + input;
    }

}
